package MulticastQuene;

import point_to_point_queue.PointToPointQueueReceiverEnd;
import point_to_point_queue.PointToPointQueueReceiverEndNonRobust;
import point_to_point_queue.PointToPointQueueSenderEnd;
import point_to_point_queue.PointToPointQueueSenderEndNonRobust;
import replicated_calculator.Parameters;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.PriorityBlockingQueue;

/**
 *
 * An implementation of MulticastQueue, which only implements the TOTAL
 * delivery guarantee.
 * @Co-author Søren Frederiksen
 * @Co-author Lars Christian Ersly
 * @author Jesper Buus Nielsen, Aarhus University, 2012.
 */
public class MulticastQueueTotalOnly_v2<E extends Serializable>
        extends Thread implements MulticastQueue<E>  {

//	********************************************************************
//	**********************         FIELDS         **********************
//	********************************************************************

    private boolean debug = Parameters.debugQueue;


    /** The address on which we listen for incoming messages. */
    private InetSocketAddress myAddress;

    /** Used to signal that the queue is leaving the peer group. */
    private boolean isLeaving = false;

    /** Used to signal that no more elements will be added to the queue of pending gets.   */
    private boolean noMoreGetsWillBeAdded = false;

    /**
     * The peers who have a connection to us. Used to make sure that
     * we do not close down the receiving end of the queue before all
     * sending to the queue is done. Not strictly needed, but nicer.
     */
    private final HashSet<InetSocketAddress> hasConnectionToUs;

    /**The incoming message queue. All other peers send their messages to this queue. */
    private PointToPointQueueReceiverEnd<MulticastMessageTotal> incoming;

    /**Keeping track of the outgoing message queues, stored under the corresponding internet address. */
    private final ConcurrentHashMap<InetSocketAddress,PointToPointQueueSenderEnd<MulticastMessageTotal>> outgoing;

    /**
     * Contains a list of the AcknowledgeMessages received but not yet handled. This occurs when they arrive before
     * their intended message.
      */
    private final ConcurrentLinkedQueue<Integer> acknowledgedMessages;

    /**Messages pending delivery.*/
    private final PriorityBlockingQueue<MulticastMessageTotal> pendingGets;

    /** Objects pending sending. */
    private final ConcurrentLinkedQueue<E> pendingSends;

    /** Contains the current clock time for the peer */
    private int clock; //Enough in our case

    private boolean allSaidGoodBye = false;

//	********************************************************************
//	**********************      CONSTRUCTORS      **********************
//	********************************************************************

    public MulticastQueueTotalOnly_v2() {
        incoming = new PointToPointQueueReceiverEndNonRobust<MulticastMessageTotal>();
        pendingGets = new PriorityBlockingQueue<MulticastMessageTotal>();
        pendingSends = new ConcurrentLinkedQueue<E>();
        outgoing = new ConcurrentHashMap<InetSocketAddress,PointToPointQueueSenderEnd<MulticastMessageTotal>>();
        /*The thread which handles outgoing traffic.  */
        SendingThread sendingThread = new SendingThread();

        hasConnectionToUs = new HashSet<InetSocketAddress>();
        acknowledgedMessages = new ConcurrentLinkedQueue<Integer>();
        clock = 0;
//        acknowledgesGotten = new ConcurrentHashMap<MulticastMessageTotal, Counter>();
        sendingThread.start();
    }

//	********************************************************************
//	**********************   INTERFACE METHODS   ***********************
//	********************************************************************		

    public void createGroup(int port, DeliveryGuarantee deliveryGuarantee)
            throws IOException {
        assertt(deliveryGuarantee==DeliveryGuarantee.NONE ||
                deliveryGuarantee==DeliveryGuarantee.TOTAL || deliveryGuarantee==DeliveryGuarantee.CAUSAL,
                 "Can at best implement TOTAL");
        // Try to listen on the given port. Exception are propagated out.
        debug("Asked to create group on port " + port);
        incoming.listenOnPort(port);

        // Record our address
        InetAddress localhost = InetAddress.getLocalHost();
        String localHostAddress = localhost.getCanonicalHostName();
        myAddress = new InetSocketAddress(localHostAddress, port);

        // Buffer a message that we have joined the group.
        addAndNotify(pendingGets,new MultiCastMessageJoinTotal(myAddress, clock));

        // Start the recieving thread.
        this.start();
    }

    public void joinGroup(int port, InetSocketAddress knownPeer,
                          DeliveryGuarantee deliveryGuarantee)
            throws IOException {
        assertt(deliveryGuarantee==DeliveryGuarantee.NONE ||
                deliveryGuarantee==DeliveryGuarantee.TOTAL || deliveryGuarantee==DeliveryGuarantee.CAUSAL,
                 "Can at best implement TOTAL");

        debug("Asked to JOIN group on address " + knownPeer + " listening on port " + port);

        // Try to listen on the given port. Exceptions are propagated
        // out of the method.
        incoming.listenOnPort(port);

        // Record our address.
        InetAddress localhost = InetAddress.getLocalHost();
        String localHostAddress = localhost.getCanonicalHostName();
        myAddress = new InetSocketAddress(localHostAddress, port);

        // Make an outgoing connection to the known peer.
        PointToPointQueueSenderEnd<MulticastMessageTotal> out = connectToPeerAt(knownPeer);
        // Send the known peer our address.
        JoinRequestMessage joinRequestMessage = new JoinRequestMessage(myAddress, clock);
        out.put(joinRequestMessage);
        // When the known peer receives the join request it will
        // connect to us, so let us remember that she has a connection
        // to us.
        hasConnectionToUs.add(knownPeer);

        // Buffer a message that we have joined the group.
        addAndNotify(pendingGets,new MultiCastMessageJoinTotal(myAddress, clock));

        // Start the receiving thread
        this.start();
    }

    public MulticastMessageTotal get() {
        debug("GET: Asked to output a message");
        // Now an object is ready in pendingObjects, unless we are
        // shutting down.

        waitForPendingGetsOrReceivedAll();

        synchronized (pendingGets) {
            if (pendingGets.isEmpty()) {
                pendingGets.notify();
                return null;
                // By contract we signal shutdown by returning null.
            } else {
                MulticastMessageTotal ret = pendingGets.poll();

                pendingGets.notify();
                if(ret instanceof JoinRelayMessage){
                    JoinRelayMessage newRet = (JoinRelayMessage) ret;
                    debug("GET: outputting JOINMESSAGE");
                    return new MultiCastMessageJoinTotal(newRet.getAddressOfJoiner(), newRet.getClock());
                }
                debug("GET: outputting MESSAGE");
                return ret;
            }
        }
    }

    /**
     * This is the receiving thread. This is just a dispatcher: it
     * will receive the incoming message and call the respective
     * handlers.
     */
    @SuppressWarnings("unchecked")
    public void run() {
        log("starting receiving thread.");
        MulticastMessageTotal msg = incoming.get();
        /* By contract we know that msg == null only occurs if
           * incoming is shut down, which we are the only ones that can
           * do, so we use that as a way to kill the receiving thread
           * when that is needed. We shut down the incoming queue when
           * it happens that we are leaving down and all peers notified
           * us that they shut down their connection to us, at which
           * point no more message will be added to the incoming
           * queue. */
        while (msg != null) {
            handle(msg);
            msg = incoming.get();
        }

        /* Before we terminate we notify callers who are blocked in
           * out get() method that no more gets will be added to the
           * buffer pendingGets. This allows them to return with a null
           * in case no message are in that buffer. */
        noMoreGetsWillBeAdded = true;
        synchronized (pendingGets) {
            pendingGets.notifyAll();
        }
        log("stopping receiving thread.");
    }

    /***
     ** The part which receives puts, buffers then and sends them.
     **/
    public void put(E object) {
        debug("PUT: asked to put message " + object + " currently outgoing is " + outgoing.keySet());
        clock++;
        synchronized(pendingSends) {
            assertt(!isLeaving,
                     "Cannot put objects after calling leaveGroup()");
            addAndNotify(pendingSends, object);
        }
        clock++;
    }

    public void leaveGroup() {
        debug("Asked to leave!");
        synchronized (pendingSends) {
            assertt(!isLeaving, "Already left the group!");
            sendToAll(new MultiCastMessageLeaveTotal(myAddress, clock));
            isLeaving =  true;
            // We wake up the sending thread. If pendingSends happen
            // to be empty now, the sending thread will know that we
            // are shutting down, so it will not starting waiting on
            // pendingSends again.
            pendingSends.notify();
        }

    }

//	********************************************************************
//	**********************    PRIVATE METHODS    ***********************
//	********************************************************************	

    /**
     * Will send a message of the given type to all peers who at some point
     * sent us a welcome message and who did not later send us a
     * goodbye message.
     */
    private void sendToAll(MulticastMessageTotal msg) {

        if (!isLeaving) {
            debug("sending " + msg + " to me");
            /* Send to self. */
            incoming.put(msg);
            /* Then send to the others. */
            sendToAllExceptMe(msg);
        }
    }

    /**
     * Sends the message to all peers except me
     * @param msg
     */
    private void sendToAllExceptMe(MulticastMessageTotal msg) {
        if (!isLeaving) {
            debug("sending " + msg + " to " + outgoing.keySet());
            for (PointToPointQueueSenderEnd<MulticastMessageTotal> out :
                    outgoing.values()) {
                out.put(msg);
            }
        }
    }
    /**
     * Sends the message to a subset of the peers.
     * @param msg
     * @param sendTo
     */
    private void sendToSet(MulticastMessageTotal msg,Set<InetSocketAddress> sendTo) {
        if (!isLeaving && sendTo != null) {
            for (InetSocketAddress address : sendTo){
                if(outgoing.keySet().contains(address)){	//In the  case it has been disconnected
                    outgoing.get(address).put(msg);
                }
            }
            if(sendTo.contains(myAddress)){
                incoming.put(msg);
            }
            
        }
    }


    /**
     *  Should contain a mapping for each message in pendingGets to how many acknowledges have been gotten for that
     *  message.
     */
 //   private Map<MulticastMessageTotal, Counter> acknowledgesGotten;

    /**
     * Used by callers to wait for objects to enter pendingGets. When
     * the method returns, then either the collection is non-empty, or
     * the multicast queue has seen its own leave message arrive on
     * the incoming stream.
     */
    private void waitForPendingGetsOrReceivedAll() {

        //We check whether the first is ready to be delivered
        if(firstIsReady()) return;

        //Else we wait a bit and try again.
        try {
            sleep(50);
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }
        waitForPendingGetsOrReceivedAll();
    }

    /**
     * Returns whether the first message in pendingGets is ready to be retrieved. By convention if the queue is
     * shutting down it return true if no more messages is left.
     * @return
     */
    private boolean firstIsReady(){
        synchronized (pendingGets) {
            if(pendingGets.isEmpty() && noMoreGetsWillBeAdded) return true;   //We return null in get by doing this, we have shut down.
            if(pendingGets.isEmpty()) return false;                            //We are not shut down, så not ready.
            //Now we know there are pending gets so we want the first to be ready.
            pendingGets.notify();
            return pendingGets.peek().hasBeenAcknowledged();
        }
    }

    /**
     * Used to add an element to a collection and wake up one thread
     * waiting for elements on the collection.
     */
    private <T> void addAndNotify(Collection<T> coll, T object) {
        synchronized (coll) {
            coll.add(object);

            // Notify that there is a new message.
            coll.notify();
        }
    }

    /**
     * Used to create an outgoing queue towards the given address,
     * including the addition of that queue to the set of queues.
     *
     * @param address The address of the peer we want to connect
     *        to. Returns null when attempting to make connection to
     *        self.
     */
    private PointToPointQueueSenderEnd<MulticastMessageTotal> connectToPeerAt(InetSocketAddress address) {
        debug("connecting To Peer!");
        assertt(!address.equals(myAddress), "Cannot connect to self.");
        // Do we have a connection already?
        PointToPointQueueSenderEnd<MulticastMessageTotal> out
                = outgoing.get(address);
        assertt(out == null, "Cannot connect twice to same peer!");
        out = new PointToPointQueueSenderEndNonRobust<MulticastMessageTotal>();
        out.setReceiver(address);
        outgoing.put(address, out);
        log(myAddress + ": connects to " + address);
        return out;
    }

    /**
     * Used by the sending thread to wait for objects to enter the
     * collection or us having left the group. When the method
     * returns, then either the collection is non-empty, or the
     * multicast queue was called in leaveGroup();
     */
    private void waitForPendingSendsOrLeaving() {
        synchronized (pendingSends) {
            while (pendingSends.isEmpty() && !isLeaving) {
                try {
                    // We will be woken up if an object arrives or we
                    // are leaving the group. Both might be the case
                    // at the same time.
                    pendingSends.wait();
                } catch (InterruptedException e) {
                    // Probably leaving. The while condition will
                    // ensure proper behavior in case of some other
                    // interruption.
                }
            }
            // Now: pendingSends is non empty or we are leaving the group.
        }
    }

    private void disconnectFrom(InetSocketAddress address) {
        synchronized (outgoing) {
            PointToPointQueueSenderEnd<MulticastMessageTotal> out
                    = outgoing.get(address);
            if (out != null) {
                outgoing.remove(address);
                out.put(new GoodbyeMessage(myAddress,clock));
                log("disconnected from " + address);
                out.shutdown();
            }
        }
    }

//	********************************************************************
//	*****************    PRIVATE METHODS -handlers    ******************
//	********************************************************************		
    private Random rand = new Random();

    private void handle(MulticastMessageTotal msg){
        if(rand.nextInt(70) > 65 && debug){
            System.out.print("l");
        }
        
        if(msg.getSentTo() != null){
            msg.setAcknowledges(msg.getSentTo().size());
        }

        log("clock incremented to max (+1) of " + clock + " and " + msg.getClock());
        clock = Math.max(clock, msg.getClock()) +1;

        //Acknowledge the message if it needs to, notice that many messages don't need to be acknowledged.
        if(!msg.hasBeenAcknowledged()){
            String to = "";
            for(InetSocketAddress add : msg.getSentTo()){
                to += add.getAddress() +":" + add.getPort() + ",";
            }
            
            debug(myAddress.getPort() + " received msg " + msg.hashCode() + " from " + msg.getSender().getPort() + " sending acknowledge to ports " + to + " the message: " + msg);
            sendToSet(new AcknowledgeMessage(myAddress , msg.hashCode() , -1), msg.getSentTo());
            
            //Check if it has already been acknowledged here?
            synchronized(acknowledgedMessages) {
                if(acknowledgedMessages.contains(msg.hashCode())){
                    debug("PRIOR: Received message has already been acknowledged (at least once)! Acknowledging");
                    Iterator<Integer> e = acknowledgedMessages.iterator();
                    while (e.hasNext()) {
                        if (msg.hashCode() == e.next()) {
                            debug("PRIOR: One acknowledge found!");
                            e.remove();
                            msg.acknowledgeGotten();
                        }
                    }
                    if(acknowledgedMessages.contains(msg.hashCode()) ){
                        debug("PRIOR-ERROR: Still have acknowledges for the message!");
                    }
                    else{
                        debug("PRIOR: Received message has now been acknowledge by the previous.");
                    }
                }

                acknowledgedMessages.notify();
            }
            
        }

        clock++;

        if(msg instanceof AcknowledgeMessage){
            AcknowledgeMessage aMsg = (AcknowledgeMessage)msg;
            handle(aMsg);
        } else if (msg instanceof MultiCastMessagePayloadTotal) { 	                //Payload
            MultiCastMessagePayloadTotal pMsg = (MultiCastMessagePayloadTotal)msg;
            handle(pMsg);
        } else if (msg instanceof JoinRequestMessage) {		                        //JoinRequest
            JoinRequestMessage jRMsg = (JoinRequestMessage)msg;
            handle(jRMsg);
        } else if (msg instanceof JoinRelayMessage) {								//JoinRelay
            JoinRelayMessage jMsg = (JoinRelayMessage)msg;
            handle(jMsg);
        } else if (msg instanceof WelcomeMessage) {			//Welcome
            WelcomeMessage wMsg = (WelcomeMessage)msg;
            handle(wMsg);
        } else if (msg instanceof MultiCastMessageLeaveTotal) {	                    //Leave
            MultiCastMessageLeaveTotal lMsg = (MultiCastMessageLeaveTotal)msg;
            handle(lMsg);
        } else if (msg instanceof GoodbyeMessage) {									//Goodbye
            GoodbyeMessage gMsg = (GoodbyeMessage)msg;
            handle(gMsg);
        }
    }

    private void handle(AcknowledgeMessage aMsg){
        debug(myAddress.getPort() + " received acknowledge for " + aMsg.getAcknowledgedMessageID() + " from " + aMsg.getSender().getPort());
        int acknowledgedMessageID = aMsg.getAcknowledgedMessageID();
        boolean isItHere = false;
//        synchronized (pendingGets){
            for(MulticastMessageTotal msg : pendingGets){
                if(msg.hashCode() == acknowledgedMessageID){
                    debug("CURRENT: Acknowledged message " + msg.hashCode());
                    msg.acknowledgeGotten();
                    isItHere = true;
                    break;
                }
            }

 //           pendingGets.notify();
 //       }
        if(!isItHere){
            synchronized(acknowledgedMessages) {
                debug("CURRENT: Message not found, saving acknowledge for later.");
                acknowledgedMessages.add(aMsg.getAcknowledgedMessageID());
                acknowledgedMessages.notify();
            }
        }
        synchronized (acknowledgedMessages) {       //Signal that something has been acknowledged.
            acknowledgedMessages.notify();
        }


    }

    /**
     * A join request message is handled by connecting to the peer who
     * wants to join and then broadcasting a join message to all peers
     * in the current group, so that they cannot connect to the new
     * peer too.
     */
    private void handle(JoinRequestMessage jRMsg) {
        log(jRMsg);
        // When the joining peer sent the join request it connected to
        // us, so let us remember that she has a connection to us.
        synchronized(hasConnectionToUs) {
            hasConnectionToUs.add(jRMsg.getSender());
        }
        // Buffer a join message so it can be gotten.

        //We modify the procedure to send a JoinRelay to all to make sure that everybody deliver it at the same time.

        
        
        sendToAll(new JoinRelayMessage(myAddress,jRMsg.getSender(),clock));
        // Then we tell the rest of the group that we have a new member.
        log("clock incremented to " + clock++);

        clock++;

        // Then we connect to the new peer.
        PointToPointQueueSenderEnd<MulticastMessageTotal> out = connectToPeerAt(jRMsg.getSender());
        out.put(new WelcomeMessage(myAddress,null,clock));
    }



    /**
     * A join message is handled by making a connection to the new
     * peer plus sending her a welcome message with our own address.
     */
    private void handle(JoinRelayMessage jMsg) {
        log(jMsg);
        assertt(!(jMsg.getAddressOfJoiner().equals(myAddress)),
                 "Got a join message about my own joining!");

        //TODO add joinRelay istedet?
        // Buffer a join message so it can be gotten.
        //addAndNotify(pendingGets,new MulticastQueueTotalOnly.MultiCastMessageJoinTotal(jMsg.getAddressOfJoiner(), jMsg.getSentTo(),clock));
        addAndNotify(pendingGets,jMsg);
        log("clock incremented to " + clock++);
        // Connect to the new peer and bid him welcome.
        if(!jMsg.getSender().equals(myAddress)){
            PointToPointQueueSenderEnd<MulticastMessageTotal> out = connectToPeerAt(jMsg.getAddressOfJoiner());
            out.put(new WelcomeMessage(myAddress,null,clock));
            synchronized(hasConnectionToUs) {
                hasConnectionToUs.add(jMsg.getAddressOfJoiner());
            }
        }

        // When this peer receives the wellcome message it will
        // connect to us, so let us remember that she has a connection
        // to us.


    }

    /**
     * A wellcome message is handled by making a connection to the
     * existing peer who sent the wellcome message. After this,
     * SendToAll will send a copy also to the peer who sent us this
     * wellcome message.
     */
    private void handle(WelcomeMessage wmsg) {
        log(wmsg);
        // When the sender sent us the wellcome message it connect to
        // us, so let us remember that she has a connection to us.
        //But we also get a welcome message from the peer we connected to
        //So lets not add her as a connection.
        boolean add = true;
        synchronized(hasConnectionToUs) {
            if(hasConnectionToUs.contains(wmsg.getSender())) {
                add = false;
            }
            else{
                hasConnectionToUs.add(wmsg.getSender());
            }

        }
        if(add){
            connectToPeerAt(wmsg.getSender());
        }

    }

    /**
     * A payload message is handled by adding it to the queue of
     * received messages, so that it can be gotten.
     */
    private void handle(MultiCastMessagePayloadTotal pmsg) {
        log(pmsg);
        addAndNotify(pendingGets,pmsg);
    }

    /**
     * A leave message is handled by removing the connection to the
     * leaving peer.
     */
    private void handle(MultiCastMessageLeaveTotal lmsg) {
        log(lmsg);
        addAndNotify(pendingGets,lmsg);
        InetSocketAddress address = lmsg.getSender();


        if (!address.equals(myAddress)) {

            disconnectFrom(address); //Handles disconnection them and sending a goodbye.
        } else {
            // That was my own leave message. If I'm the only one left
            // in the group, then this means that I can safely shut
            // down.
            if (hasConnectionToUs.isEmpty()) {
                incoming.shutdown();
            }
        }
    }

    /**
     * A goodbuy message is produced as response to a leave message
     * and is handled by closing the connection to the existing peer
     * who sent the goodbuy message. After this, SendToAll will not
     * send a copy to the peer who sent us this goodbuy message.
     */
    private void handle(GoodbyeMessage gmsg) {
        log(gmsg);
        // When the peer sent us the goodbuy message, it closed its
        // connection to us, so let us remember that.
        synchronized(hasConnectionToUs) {
            hasConnectionToUs.remove(gmsg.getSender());
            log("now " + hasConnectionToUs.size() + " has connections to us!");
            // If we are leaving and that was the last goodbuy
            // message, then we can shut down the incoming queue and
            // terminate the receving thread.
            if (hasConnectionToUs.isEmpty() && isLeaving) {
                // If the receiving thread is blocked on the incoming
                // queue, it will be woken up and receive a null when
                // the queue is empty, which will tell it that we have
                // received all messages.

                incoming.shutdown();
            }
        }
    }



//	********************************************************************
//	*********************  HELPERS FOR DEBUGGING  **********************
//	********************************************************************

    protected boolean log = false;

    public void printLog() {
        log = true;
    }

    protected void log(String msg) {
        if (log) System.out.println("MCQ: " + myAddress + " multicastQueue said: " + msg);
    }

    protected void log(MulticastMessage msg) {
        if (log) System.out.println("MCQ: " + myAddress + " multicastQueue received: " + msg);

    }

//	********************************************************************
//	***************    INTERNAL CLASSES -SendThread    *****************
//	********************************************************************	


    /**
     * Will take objects from pendingSends and send them to all peers.
     * If the queue empties and leaveGroup() was called, then the
     * queue will remain empty, so we can terminate.
     */
    private class SendingThread extends Thread {
        public void run() {
            log("starting sending thread.");
            // As long as we are not leaving or there are objects to
            // send, we will send them.
            waitForPendingSendsOrLeaving();
            E object = pendingSends.poll();
            while (object != null) {
                Set<InetSocketAddress> all = new HashSet<InetSocketAddress>();
                all.add(myAddress);
                all.addAll(outgoing.keySet());
                sendToAll(new MultiCastMessagePayloadTotal<E>(myAddress, all, object, clock));
                waitForPendingSendsOrLeaving();
                object = pendingSends.poll();
            }
            log("shutting down outgoing connections.");
            synchronized (outgoing) {
                for (InetSocketAddress address : outgoing.keySet()) {
                    disconnectFrom(address);
                }
            }
            log("stopping sending thread.");
        }
    }

    private void debug(String s){
        if(debug) System.out.println("MCQ: " + s);
    }
    
    private void assertt(boolean b, String s){
        if(!b){
            System.out.println(s);
            System.exit(-1);
        }
    }
}

